package com.hitqz.robot.biz.websocket.endpoint;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.hitqz.robot.api.common.enums.SubscribeTopicEnum;
import com.hitqz.robot.biz.websocket.config.CustomConfigurator;
import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;
import jakarta.websocket.server.ServerEndpoint;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Arrays;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

@ServerEndpoint(value = "/subscribe/{topic}",configurator = CustomConfigurator.class)
@Component
@Slf4j
public class SubscriptionEndpoint {

	private static final ObjectMapper objectMapper = new ObjectMapper();
	private static final Map<String, CopyOnWriteArraySet<Session>> topicSubscribers = new ConcurrentHashMap<>();
	private static final Set<Session> sessions = new CopyOnWriteArraySet<>();

	@OnOpen
	public void onOpen(Session session,@PathParam(value = "topic") String topic) throws IOException {
		try {
			SubscribeTopicEnum subscribeTopicEnum = Arrays.stream(SubscribeTopicEnum.values()).filter(
					t-> t.getTopic().equals(topic)).findFirst().get();
			sessions.add(session);
			subscribe(session, subscribeTopicEnum.getTopic());
		}catch (Exception e) {
			if (log.isWarnEnabled()) {
				session.getBasicRemote().sendText("无法解析该话题信息");
				log.warn("话题关注发生异常无法解析 {}", topic);
			}
		}
	}

	@OnMessage
	public void onMessage(Session session,String message) {
		//log.info("[ws][subscribe],message:{},session-id:{}", message,session.getId());
		// 如果是设备的话需要关注


	}

	@OnClose
	public void onClose(Session session,@PathParam(value = "topic") String topic) {
		//取消订阅
		unsubscribe(session, topic);
		// 移除session
		sessions.remove(session);
		log.debug("[ws][subscribe],disconnected.current size :{},session-id:{}",sessions.size(),session.getId());
	}

	@OnError
	public void onError(Session session,Throwable error) {
		log.error("[ws][subscribe],error:{},session-id:{}",error.getMessage(),session.getId());
	}

	// 话题广播
	public static void broadcast(String topic, String message) {
		if (topicSubscribers.containsKey(topic)) {
			if (topicSubscribers.get(topic).isEmpty()) {
				topicSubscribers.remove(topic);
				return;
			}
			for (Session subscriber : topicSubscribers.get(topic)) {
				if (subscriber.isOpen()) {
					subscriber.getAsyncRemote().sendText(message);
				}
			}
		}
	}

	// 订阅话题
	public static void subscribe(Session session, String topic) {
		topicSubscribers.computeIfAbsent(topic, k -> new CopyOnWriteArraySet<>()).add(session);
		log.info("[ws][subscribe],subscribe topic:{},session-id:{}",topic,session.getId());
	}

	// 取消订阅
	public static void unsubscribe(Session session, String topic) {
		// 移除订阅session
		if (topicSubscribers.containsKey(topic)) {
			topicSubscribers.get(topic).remove(session);
			// 进一步移除话题
			if (topicSubscribers.get(topic).isEmpty()) {
				topicSubscribers.remove(topic);
			}
			log.info("[ws][subscribe],unsubscribe topic:{},session-id:{}",topic,session.getId());
		}
	}

	private void sendError(Session session, String errorMessage) {
		try {
			session.getBasicRemote().sendText("{\"error\":\"" + errorMessage + "\"}");
		} catch (IOException e) {
			// 处理发送错误消息时的异常
		}finally {
			try {
				// 在发送错误信息后关闭会话
				session.close(new CloseReason(CloseReason.CloseCodes.CANNOT_ACCEPT, errorMessage));
			} catch (IOException e) {
				// 处理关闭session时的异常
				if (log.isErrorEnabled()) {
					log.error("[ws][subscribe]close, session-id:{} failed:{}",session.getId(),e.getMessage());
				}
			}
		}
	}

}
